package com.yet.container.remote.netty.client;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yet.container.registry.RegistryMeta.Address;
import com.yet.container.remote.ResponseFuture;
import com.yet.dto.Request;
import com.yet.dto.Response;
import com.yet.exception.RemotingException;
import com.yet.serialization.Serialization;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;

public class NettyClientProxy {

	private static final Logger logger = LoggerFactory.getLogger(NettyClientProxy.class);

	public final InetSocketAddress socketAddress;

	private Serialization serialization;

	final EventLoopGroup group = new NioEventLoopGroup();

	private Channel channel;

	public NettyClientProxy(Address address, Serialization serialization) {
		super();
		this.socketAddress = new InetSocketAddress(address.getHost(), address.getPort());
		this.serialization = serialization;
	}

	public void initChannel() throws InterruptedException {
		// 客户端
		Bootstrap bootstrap = new Bootstrap();

		bootstrap.group(group).channel(NioSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO))
				.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true);
		ChannelHandler[] handlers = new ChannelHandler[] { new IdleStateHandler(0, 1, 0, TimeUnit.MINUTES),
				new NettyDecoder(Response.class), new NettyEncoder(), new NettyClientHandler() };

		bootstrap.handler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(handlers);
			}
		});
		channel = bootstrap.connect(socketAddress).sync().addListener(f -> {
			if (f.isSuccess()) {
				logger.debug("Connect {} success", socketAddress);
			} else {
				logger.debug("Connect {} failed", socketAddress);
			}
		}).channel();
	}

	public void close() {
		group.shutdownGracefully(0, 5000, TimeUnit.SECONDS);
		if (channel != null && channel.isOpen()) {
			channel.close();
		}
	}

	public void send(Request request) throws RemotingException {
		this.channel.writeAndFlush(request);
	}

	class NettyDecoder extends ByteToMessageDecoder {

		private Class<?> clzz;

		public NettyDecoder(Class<?> clzz) {
			this.clzz = clzz;
		}

		@Override
		protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
			if (in.readableBytes() < 4) {
				return;
			}
			in.markReaderIndex();
			int dataLength = in.readInt();
			if (dataLength < 0) {
				ctx.close();
			}
			if (in.readableBytes() < dataLength) {
				in.resetReaderIndex();
				return; // fix 1024k buffer splice limix
			}
			byte[] data = new byte[dataLength];
			in.readBytes(data);

			Object obj = serialization.deserialize(data, clzz);
			out.add(obj);
		}
	}

	class NettyEncoder extends MessageToByteEncoder<Request> {

		@Override
		protected void encode(ChannelHandlerContext ctx, Request msg, ByteBuf out) throws Exception {
			byte[] data = serialization.serialize(msg);
			out.writeInt(data.length);
			out.writeBytes(data);
		}
	}

	@ChannelHandler.Sharable
	class NettyClientHandler extends ChannelInboundHandlerAdapter {

		private final ByteBuf HEARTBEAT_SEQUENCE = Unpooled
				.unreleasableBuffer(Unpooled.copiedBuffer("Heartbeat", CharsetUtil.UTF_8));

		@Override
		public void channelActive(ChannelHandlerContext ctx) throws Exception {
			super.channelActive(ctx);
		}

		@Override
		public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
			if (msg instanceof Response) {
				ResponseFuture.received((Response) msg);
			} else {
				logger.error("Server exception with error responseType :{}", msg.getClass());
			}
		}

		@Override
		public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
			if (evt instanceof IdleStateEvent) {
				IdleState state = ((IdleStateEvent) evt).state();
				if (state == IdleState.WRITER_IDLE) {
					logger.debug("ping");
					ctx.writeAndFlush(HEARTBEAT_SEQUENCE.duplicate());
				}
			} else {
				super.userEventTriggered(ctx, evt);
			}
		}
	}
}
